package com.zhouyu.fileSync.FSclient.registerClient;

import com.zhouyu.fileSync.FScommon.api.JSONDecoder;
import com.zhouyu.fileSync.FScommon.api.JSONEncoder;
import com.zhouyu.fileSync.FScommon.api.RpcNettyClient;
import com.zhouyu.fileSync.FScommon.api.entities.Request;
import com.zhouyu.fileSync.FScommon.api.entities.Response;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.timeout.IdleStateHandler;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.concurrent.SynchronousQueue;

/**
 * register client to connect to register server
 * and send ,receive message through netty channel
 */
public class RegisterClient implements RpcNettyClient {
    Logger logger = LogManager.getLogger(RegisterClient.class);
    private Channel channel;
    private RegisterClientHandler clientHandler = new RegisterClientHandler();
    private String host;
    private int port;

    public RegisterClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    @Override
    public void run() {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group).channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .option(ChannelOption.TCP_NODELAY, true).handler(new ChannelInitializer<SocketChannel>() {

                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    ChannelPipeline pipeline = socketChannel.pipeline();
                    pipeline.addLast(new IdleStateHandler(0, 0, 10));
                    pipeline.addLast(new DelimiterBasedFrameDecoder(65535, Unpooled.copiedBuffer("#_@_#".getBytes())));
                    pipeline.addLast(new JSONEncoder());
                    pipeline.addLast(new JSONDecoder());
                    pipeline.addLast(clientHandler);
                }
            });

            ChannelFuture cf = bootstrap.connect(host, port).sync();
            logger.info("netty rpc client start up! connect to host:" + host + ":" + port);
            //等待服务端监听端口关闭
            channel = cf.channel();
            channel.closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
            logger.error(e);
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * send request to server
     *
     * @param request
     * @return
     */
    @Override
    public Response sendToServer(Request request) throws InterruptedException {
        if (channel != null && channel.isActive()) {
            SynchronousQueue<Object> queue = clientHandler.sendRequest(request, channel);
            Object response = queue.take();
            return (Response) response;
        } else {
            Response response = new Response();
            response.setCode(1);
            response.setErrorMsg("failed to connect to server,please check the configs!");
            return response;
        }

    }

    public Channel getChannel() {
        return channel;
    }

}
